#lang racket/base
(require racket/set
		 racket/contract
		 racket/function
		 "definition.rkt"
		 "utility.rkt"
		 "subscriber.rkt")
(provide (all-defined-out))

(define/contract (create-subject)
  (-> subject?)
  (define sbj
	(subject (lambda (sub)
			   ;; 添加当前subscriber
			   (set-add! (subject-subscribers sbj) sub)
			   (disposable #f (mutable-seteq)
						   (lambda (dsp)
							 ;; dispose时，移除改subscriber
							 (set-remove! (subject-subscribers sbj) sub))))
			 (mutable-seteq)))
  sbj)

(define/contract (send-next sbj value)
  (-> subject? any/c any)
  (for-each (curryr on-next value) (set->list (subject-subscribers sbj))))

(define/contract (send-error sbj value)
  (-> subject? error? any)
  (for-each (curryr on-error value) (set->list (subject-subscribers sbj))))

(define/contract (send-completed sbj)
  (-> subject? any)
  (for-each on-completed (set->list (subject-subscribers sbj))))

(module+ test
  (define sbj (create-subject))
  (subscribe sbj
			 (lambda (x) (displayln x))
			 (lambda (e) (displayln "error"))
			 (lambda () (displayln "completed")))
  (send-next sbj 1)
  (send-next sbj 2)
  (send-next sbj 3)
  (send-error sbj (error "error" #f))
  (send-completed sbj))
